#include "EventLoop.h"
#include "Logger.h"
#include "Poller.h"
#include "Channel.h"

#include <sys/eventfd.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>

//防止一個線程創建多個EventLoop __thread這是thread local，不然就是一個全局變量。無法讓每一個線程獨享
__thread EventLoop *t_loopInThisThread = nullptr;

//定義默認的Poller IO複用接口的超時時間（10S）
const int kPollTimeMS = 10000;

//創建wakeupfd，用來 notify subReactor新來的channel
int createEventFd(){
    int eventFd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if(eventFd < 0){
        LOG_FATAL("func = %s, eventfd error : %d \n", _FUNCTION_, errno);
    }
    return eventFd;
}

EventLoop::EventLoop() : looping_(false), quit_(false), callingPendingFunctors_(false)
        , threadId_(CurrentThread::tid()), poller_(Poller::newDefaultPoller(this))
        , wakeupFd_(createEventFd()), wakeupChannel_(new Channel(this, wakeupFd_))
        , currentActiveChannel_(nullptr){
    LOG_DEBUG("EventLoop created %p in thread %d \n", this, threadId_);
    if(t_loopInThisThread){
        LOG_FATAL("Another EventLoop %p exists in this thread %d \n", t_loopInThisThread, threadId_);
    }
    else{
        t_loopInThisThread = this;
    }

    //設置wakefd的事件類型以及發生事件後的回調操作
    wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead, this));
    //每一個eventLoop都將監聽wakeupChannel的EPOLLIN事件
    wakeupChannel_->enableReading();
}

EventLoop::~EventLoop(){
    wakeupChannel_->disableAll();
    wakeupChannel_->remove();
    ::close(wakeupFd_);
    t_loopInThisThread = nullptr;
}

void EventLoop::handleRead(){
    uint64_t one = 1;
    ssize_t n = read(wakeupFd_, &one, sizeof(one));
    if(n != sizeof(one)){
        LOG_ERROR("EventLoop::handleRead() reads %d bytes instead of 8", n);
    }
}

void EventLoop::loop(){
    looping_ = true;
    quit_ = false;
    LOG_INFO("EventLoop %p start looping \n", this);
    while(!quit_){
        activeChannels_.clear();
        //监听两类fd，一种是client的fd，一种是wakeupfd（mainLoop和subLoop通信的fd）
        pollReturnTime_ = poller_->poll(kPollTimeMS, &activeChannels_);
        for(Channel *channel : activeChannels_){
            //Poller监听那些channel发生事件，然后上报给EventLoop，通知channel处理相应的事件
            channel->handleEvent(pollReturnTime_);
        }
        //执行当前EventLoop事件循环需要处理的回调函数
        /*
            IO线程 mainLoop accept fd => subLoop
            mainLoop 事先注册一个回调函数cb 需要subLoop来执行 通过wakeup唤醒，执行之前mainLoop注册的一个或者多个回调函数
        */
       doPendingFunctors();
    }
    LOG_INFO("EventLoop %p stop looping \n", this);
    looping_ = false;
}

void EventLoop::quit(){
    /*
        1.loop在自己的线程中调用自己的quit
            quit = true >>> 由于是在自己线程，所以poll是已经被唤醒了，
                此时会去判断while条件，发现不成立，所以自然退出while循环
        2.loop不在自己线程中调用quit，也就是在其他线程中调用quit，比如在subLoop中调用mainLoop中quit
            quit = true >>> 由于不是在自己线程，要关闭别的loop（现在不知道别个loop是否阻塞），所以应该是先将别人唤醒，
            如果此时别个loop在阻塞中（poll），先解放出来，然后去判断while循环条件。如果此时别个loop
            正在忙碌中，那就是等它处理完后，然后阻塞在pool中，再将之唤醒。
    */
    quit_ = true;
    if(!isInLoopThread()){
        wakeup();   //唤醒别个loop（阻塞在poll）
    }
}

// 在当前loop中执行cb
void EventLoop::runInLoop(Functor cb){
    if (isInLoopThread()) // 当前EventLoop中执行回调
    {
        cb();
    }
    else // 在非当前EventLoop线程中执行cb，就需要唤醒EventLoop所在线程执行cb
    {
        queueInLoop(cb);
    }
}

// 把cb放入队列中 唤醒loop所在的线程执行cb
void EventLoop::queueInLoop(Functor cb){
    {
        std::unique_lock<std::mutex> lock(mutex_);
        pendingFunctors_.emplace_back(cb);
    }

    /**
     * || callingPendingFunctors的意思是 当前loop正在执行回调中 但是loop的pendingFunctors_中又加入了新的回调 需要通过wakeup写事件
     * 唤醒相应的需要执行上面回调操作的loop的线程 让loop()下一次poller_->poll()不再阻塞（阻塞的话会延迟前一次新加入的回调的执行），然后
     * 继续执行pendingFunctors_中的回调函数
     **/
    if (!isInLoopThread() || callingPendingFunctors_){
        wakeup(); // 唤醒loop所在线程
    }
}

// EventLoop的方法 => Poller的方法
void EventLoop::updateChannel(Channel *channel)
{
    poller_->updateChannel(channel);
}

void EventLoop::removeChannel(Channel *channel)
{
    poller_->removeChannel(channel);
}

bool EventLoop::hasChannel(Channel *channel)
{
    return poller_->hasChannel(channel);
}

// 用来唤醒loop所在线程 向wakeupFd_写一个数据 wakeupChannel就发生读事件 当前loop线程就会被唤醒
void EventLoop::wakeup()
{
    uint64_t one = 1;
    ssize_t n = write(wakeupFd_, &one, sizeof(one));
    if (n != sizeof(one))
    {
        LOG_ERROR("EventLoop::wakeup() writes %lu bytes instead of 8\n", n);
    }
}

void EventLoop::doPendingFunctors(){
    std::vector<Functor> functors;
    callingPendingFunctors_ = true;

    {
        std::unique_lock<std::mutex> lock(mutex_);
        functors.swap(pendingFunctors_); // 交换的方式减少了锁的临界区范围 提升效率 同时避免了死锁 如果执行functor()在临界区内 且functor()中调用queueInLoop()就会产生死锁
    }

    for (const Functor &functor : functors){
        functor(); // 执行当前loop需要执行的回调操作
    }

    callingPendingFunctors_ = false;
}